From 7f2d4d2a6fed5f54808516f34570b7a1aeddd39f Mon Sep 17 00:00:00 2001 From: phernandez Date: Mon, 2 Mar 2026 12:58:22 -0600 Subject: [PATCH] fix: three cloud-testing bugs (#640, #641, #642) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 🔧 #640 — LinkResolver selects worst match instead of best Replace `min(results, key=lambda x: x.score)` with `results[0]`. Both SQLite and Postgres return results sorted best-first in SQL, so using `results[0]` is backend-agnostic and correct. 🔧 #641 — search_notes output_format="text" returns raw Pydantic model Add `_format_search_markdown()` that formats SearchResponse as readable markdown with title, permalink, score, and matched snippet per result. Update prompts to use `output_format="json"` since they need structured data for result counting and branching logic. 🔧 #642 — metadata_filters with `note_type` key returns empty results Add `_METADATA_KEY_ALIASES` mapping at the tool level that aliases `note_type` → `type` before passing metadata_filters to the search query. The frontmatter field is `type`, not `note_type`. Co-Authored-By: Claude Opus 4.6 Signed-off-by: phernandez --- .../mcp/prompts/continue_conversation.py | 29 +- src/basic_memory/mcp/prompts/search.py | 40 +- src/basic_memory/mcp/tools/search.py | 54 ++- src/basic_memory/services/link_resolver.py | 6 +- test-int/mcp/test_delete_note_integration.py | 14 +- test-int/mcp/test_search_integration.py | 11 +- tests/mcp/test_prompt_tool_delegation.py | 57 +-- tests/mcp/test_tool_search.py | 344 +++++++++++++++--- tests/services/test_link_resolver.py | 18 + 9 files changed, 431 insertions(+), 142 deletions(-) diff --git a/src/basic_memory/mcp/prompts/continue_conversation.py b/src/basic_memory/mcp/prompts/continue_conversation.py index a4594a89..9020a743 100644 --- a/src/basic_memory/mcp/prompts/continue_conversation.py +++ b/src/basic_memory/mcp/prompts/continue_conversation.py @@ -13,7 +13,6 @@ from pydantic import Field from basic_memory.mcp.server import mcp from basic_memory.mcp.tools.recent_activity import recent_activity from basic_memory.mcp.tools.search import search_notes -from basic_memory.schemas.search import SearchResponse @mcp.prompt( @@ -42,15 +41,12 @@ async def continue_conversation( logger.info(f"Continuing session, topic: {topic}, timeframe: {timeframe}") if topic: - # Search for the topic using the search tool directly - result = await search_notes(query=topic, after_date=timeframe) + # Use json format to get structured data for result counting and branching + result = await search_notes(query=topic, after_date=timeframe, output_format="json") - if isinstance(result, SearchResponse): - context_text = _format_continuation_results(result, topic) - result_count = len(result.results) - elif isinstance(result, dict): + if isinstance(result, dict): results = result.get("results", []) - context_text = str(result) + context_text = _format_continuation_results(results, topic) result_count = len(results) else: # Error string @@ -111,23 +107,24 @@ async def continue_conversation( return prompt -def _format_continuation_results(result: SearchResponse, topic: str) -> str: - """Format search results for conversation continuation context.""" - if not result.results: +def _format_continuation_results(results: list[dict], topic: str) -> str: + """Format search result dicts for conversation continuation context.""" + if not results: return f"No previous context found for '{topic}'." lines = [f"## Previous Context for '{topic}'\n"] - for item in result.results: - title = item.title or "Untitled" - permalink = item.permalink or "" + for item in results: + title = item.get("title", "Untitled") + permalink = item.get("permalink", "") lines.append(f"### {title}") if permalink: lines.append(f"permalink: {permalink}") lines.append(f'Read with: `read_note("{permalink}")`') - if item.content: - content = item.content[:300] + "..." if len(item.content) > 300 else item.content + content = item.get("content") + if content: + content = content[:300] + "..." if len(content) > 300 else content lines.append(f"\n{content}") lines.append("") diff --git a/src/basic_memory/mcp/prompts/search.py b/src/basic_memory/mcp/prompts/search.py index 6c055f57..8038efd4 100644 --- a/src/basic_memory/mcp/prompts/search.py +++ b/src/basic_memory/mcp/prompts/search.py @@ -11,7 +11,6 @@ from pydantic import Field from basic_memory.mcp.server import mcp from basic_memory.mcp.tools.search import search_notes -from basic_memory.schemas.search import SearchResponse @mcp.prompt( @@ -39,18 +38,14 @@ async def search_prompt( """ logger.info(f"Searching knowledge base, query: {query}, timeframe: {timeframe}") - # Call the search tool directly — it returns SearchResponse, dict, or error string - result = await search_notes(query=query, after_date=timeframe) + # Use json format to get structured data for result counting and formatting + result = await search_notes(query=query, after_date=timeframe, output_format="json") # Format the tool output into a prompt with guidance - if isinstance(result, SearchResponse): - result_count = len(result.results) - result_text = _format_search_results(result, query) - elif isinstance(result, dict): - # json output format + if isinstance(result, dict): results = result.get("results", []) result_count = len(results) - result_text = str(result) + result_text = _format_search_results(results, query) else: # Error string from search tool result_count = 0 @@ -76,28 +71,27 @@ async def search_prompt( """) -def _format_search_results(result: SearchResponse, query: str) -> str: - """Format SearchResponse into readable markdown.""" - if not result.results: +def _format_search_results(results: list[dict], query: str) -> str: + """Format search result dicts into readable markdown.""" + if not results: return f"No results found for '{query}'." - lines = [f"Found {len(result.results)} results:\n"] + lines = [f"Found {len(results)} results:\n"] - for item in result.results: - title = item.title or "Untitled" - permalink = item.permalink or "" - score = f" (score: {item.score:.2f})" if item.score else "" + for item in results: + title = item.get("title", "Untitled") + permalink = item.get("permalink", "") + score = item.get("score") + score_text = f" (score: {score:.2f})" if score else "" - lines.append(f"- **{title}**{score}") + lines.append(f"- **{title}**{score_text}") if permalink: lines.append(f" permalink: {permalink}") - if item.content: + content = item.get("content") + if content: # Truncate content snippet - content = item.content[:200] + "..." if len(item.content) > 200 else item.content + content = content[:200] + "..." if len(content) > 200 else content lines.append(f" {content}") lines.append("") - if result.has_more: - lines.append("*More results available. Use page=2 to see next page.*") - return "\n".join(lines) diff --git a/src/basic_memory/mcp/tools/search.py b/src/basic_memory/mcp/tools/search.py index 4ff460c6..3694a7b0 100644 --- a/src/basic_memory/mcp/tools/search.py +++ b/src/basic_memory/mcp/tools/search.py @@ -251,6 +251,46 @@ Error searching for '{query}': {error_message} - **Patterns**: `tag:example`, `category:observation`""" +def _format_search_markdown(result: SearchResponse, project: str, query: str | None) -> str: + """Format SearchResponse as compact markdown text. + + Produces a human-readable markdown representation suitable for LLM + consumption when structured data isn't needed. + """ + if not result.results: + return f"No results found for '{query or ''}' in project '{project}'." + + parts = [] + + # --- Header --- + if query: + parts.append(f"# Search Results: {query}") + else: + parts.append("# Search Results") + parts.append(f"*project: {project}*") + parts.append("") + + # --- Result blocks --- + for r in result.results: + parts.append(f"### {r.title}") + parts.append(f"- permalink: {r.permalink}") + parts.append(f"- score: {r.score:.4f}") + if r.matched_chunk: + parts.append(f"- match: {r.matched_chunk[:200]}") + parts.append("") + + # --- Footer with pagination --- + parts.append("---") + count = len(result.results) + parts.append( + f"*{count} result{'s' if count != 1 else ''}" + f" | page {result.current_page}, page_size {result.page_size}" + f"{' | more available' if result.has_more else ''}*" + ) + + return "\n".join(parts) + + @mcp.tool( description="Search across all content in the knowledge base with advanced syntax support.", # TODO: re-enable once MCP client rendering is working @@ -273,7 +313,7 @@ async def search_notes( status: Optional[str] = None, min_similarity: Optional[float] = None, context: Context | None = None, -) -> SearchResponse | dict | str: +) -> dict | str: """Search across all content in the knowledge base with comprehensive syntax support. This tool searches the knowledge base using full-text search, pattern matching, @@ -373,7 +413,8 @@ async def search_notes( context: Optional FastMCP context for performance caching. Returns: - SearchResponse with results and pagination info, or helpful error guidance if search fails + Formatted markdown text (output_format="text"), dict (output_format="json"), + or helpful error guidance string if search fails Examples: # Basic text search @@ -519,6 +560,13 @@ async def search_notes( if after_date: search_query.after_date = after_date if metadata_filters: + # Alias common column/model names to their frontmatter key equivalents. + # Users often pass "note_type" (the entity model column) when the + # frontmatter field is actually "type". + _METADATA_KEY_ALIASES = {"note_type": "type"} + metadata_filters = { + _METADATA_KEY_ALIASES.get(k, k): v for k, v in metadata_filters.items() + } search_query.metadata_filters = metadata_filters if tags: search_query.tags = tags @@ -565,7 +613,7 @@ async def search_notes( if output_format == "json": return result.model_dump(mode="json", exclude_none=True) - return result + return _format_search_markdown(result, active_project.name, query) except Exception as e: logger.error( diff --git a/src/basic_memory/services/link_resolver.py b/src/basic_memory/services/link_resolver.py index 526afa55..739e0087 100644 --- a/src/basic_memory/services/link_resolver.py +++ b/src/basic_memory/services/link_resolver.py @@ -301,8 +301,10 @@ class LinkResolver: ) if results: - # Look for best match - best_match = min(results, key=lambda x: x.score) # pyright: ignore + # Both SQLite and Postgres return results sorted best-first in SQL + # (SQLite: ORDER BY score ASC for negative BM25, Postgres: ORDER BY score DESC + # for positive ts_rank). Using results[0] is backend-agnostic and correct. + best_match = results[0] logger.trace( f"Selected best match from {len(results)} results: {best_match.permalink}" ) diff --git a/test-int/mcp/test_delete_note_integration.py b/test-int/mcp/test_delete_note_integration.py index 20a99f0b..4aa16a28 100644 --- a/test-int/mcp/test_delete_note_integration.py +++ b/test-int/mcp/test_delete_note_integration.py @@ -105,11 +105,8 @@ async def test_delete_note_by_permalink(mcp_server, app, test_project): }, ) - # Should have no results - assert ( - '"results": []' in search_result.content[0].text - or '"results":[]' in search_result.content[0].text - ) + # Default text format returns "No results found" when empty + assert "No results found" in search_result.content[0].text @pytest.mark.asyncio @@ -387,11 +384,8 @@ async def test_delete_multiple_notes_sequentially(mcp_server, app, test_project) }, ) - # Should have no results - assert ( - '"results": []' in search_result.content[0].text - or '"results":[]' in search_result.content[0].text - ) + # Default text format returns "No results found" when empty + assert "No results found" in search_result.content[0].text @pytest.mark.asyncio diff --git a/test-int/mcp/test_search_integration.py b/test-int/mcp/test_search_integration.py index 99d1dd0e..1ac16e30 100644 --- a/test-int/mcp/test_search_integration.py +++ b/test-int/mcp/test_search_integration.py @@ -362,9 +362,9 @@ async def test_search_pagination(mcp_server, app, test_project): ) result_text = search_result.content[0].text - # Should contain 5 results and pagination info - assert '"current_page":1' in result_text - assert '"page_size":5' in result_text + # Text format includes pagination info in footer + assert "page 1" in result_text + assert "page_size 5" in result_text # Search page 2 search_result = await client.call_tool( @@ -378,7 +378,7 @@ async def test_search_pagination(mcp_server, app, test_project): ) result_text = search_result.content[0].text - assert '"current_page":2' in result_text + assert "page 2" in result_text @pytest.mark.asyncio @@ -407,8 +407,9 @@ async def test_search_no_results(mcp_server, app, test_project): }, ) + # Default text format returns "No results found" when empty result_text = search_result.content[0].text - assert '"results": []' in result_text or '"results":[]' in result_text + assert "No results found" in result_text @pytest.mark.asyncio diff --git a/tests/mcp/test_prompt_tool_delegation.py b/tests/mcp/test_prompt_tool_delegation.py index 8aacb37a..7718d21f 100644 --- a/tests/mcp/test_prompt_tool_delegation.py +++ b/tests/mcp/test_prompt_tool_delegation.py @@ -9,7 +9,6 @@ import pytest from basic_memory.mcp.prompts.search import search_prompt from basic_memory.mcp.prompts.continue_conversation import continue_conversation -from basic_memory.schemas.search import SearchResponse, SearchResult # --- search_prompt --- @@ -20,19 +19,20 @@ async def test_search_prompt_delegates_to_search_notes(monkeypatch): """Search prompt should call search_notes tool and wrap output.""" captured_kwargs = {} - fake_result = SearchResponse( - results=[ - SearchResult( - type="entity", - title="Test Note", - permalink="test-note", - file_path="test-note.md", - score=0.95, - ) + # Prompts use output_format="json", so mock returns a dict + fake_result = { + "results": [ + { + "type": "entity", + "title": "Test Note", + "permalink": "test-note", + "file_path": "test-note.md", + "score": 0.95, + } ], - current_page=1, - page_size=10, - ) + "current_page": 1, + "page_size": 10, + } async def fake_search_notes(**kwargs): captured_kwargs.update(kwargs) @@ -45,6 +45,7 @@ async def test_search_prompt_delegates_to_search_notes(monkeypatch): # Verify delegation assert captured_kwargs["query"] == "my query" assert captured_kwargs["after_date"] == "1w" + assert captured_kwargs["output_format"] == "json" # Verify output wrapping assert 'Search Results: "my query"' in out @@ -55,7 +56,7 @@ async def test_search_prompt_delegates_to_search_notes(monkeypatch): @pytest.mark.asyncio async def test_search_prompt_handles_no_results(monkeypatch): """Search prompt should handle empty results gracefully.""" - fake_result = SearchResponse(results=[], current_page=1, page_size=10) + fake_result = {"results": [], "current_page": 1, "page_size": 10} async def fake_search_notes(**kwargs): return fake_result @@ -91,19 +92,20 @@ async def test_continue_conversation_delegates_to_search_notes(monkeypatch): """Continue conversation with topic should call search_notes.""" captured_kwargs = {} - fake_result = SearchResponse( - results=[ - SearchResult( - type="entity", - title="Previous Discussion", - permalink="discussions/previous", - file_path="discussions/previous.md", - score=0.9, - ) + # Prompts use output_format="json", so mock returns a dict + fake_result = { + "results": [ + { + "type": "entity", + "title": "Previous Discussion", + "permalink": "discussions/previous", + "file_path": "discussions/previous.md", + "score": 0.9, + } ], - current_page=1, - page_size=10, - ) + "current_page": 1, + "page_size": 10, + } async def fake_search_notes(**kwargs): captured_kwargs.update(kwargs) @@ -117,6 +119,7 @@ async def test_continue_conversation_delegates_to_search_notes(monkeypatch): assert captured_kwargs["query"] == "my topic" assert captured_kwargs["after_date"] == "3d" + assert captured_kwargs["output_format"] == "json" assert "'my topic'" in out assert "Previous Discussion" in out @@ -164,7 +167,7 @@ async def test_continue_conversation_no_topic_default_timeframe(monkeypatch): @pytest.mark.asyncio async def test_continue_conversation_no_results_for_topic(monkeypatch): """Continue conversation should show capture opportunity when no results found.""" - fake_result = SearchResponse(results=[], current_page=1, page_size=10) + fake_result = {"results": [], "current_page": 1, "page_size": 10} async def fake_search_notes(**kwargs): return fake_result diff --git a/tests/mcp/test_tool_search.py b/tests/mcp/test_tool_search.py index 6ed804bd..cae457ab 100644 --- a/tests/mcp/test_tool_search.py +++ b/tests/mcp/test_tool_search.py @@ -5,7 +5,11 @@ from contextlib import asynccontextmanager from datetime import datetime, timedelta from basic_memory.mcp.tools import write_note -from basic_memory.mcp.tools.search import search_notes, _format_search_error_response +from basic_memory.mcp.tools.search import ( + search_notes, + _format_search_error_response, + _format_search_markdown, +) from basic_memory.schemas.search import SearchResponse @@ -22,15 +26,18 @@ async def test_search_text(client, test_project): ) assert result - # Search for it - response = await search_notes(project=test_project.name, query="searchable") + # Search for it (use json format to inspect structured results) + response = await search_notes( + project=test_project.name, query="searchable", output_format="json" + ) # Verify results - handle both success and error cases - if isinstance(response, SearchResponse): - # Success case - verify SearchResponse - assert len(response.results) > 0 + if isinstance(response, dict): + # Success case - verify dict response + assert len(response["results"]) > 0 assert any( - r.permalink == f"{test_project.name}/test/test-search-note" for r in response.results + r["permalink"] == f"{test_project.name}/test/test-search-note" + for r in response["results"] ) else: # If search failed and returned error message, test should fail with informative message @@ -50,21 +57,22 @@ async def test_search_title(client, test_project): ) assert result - # Search for it + # Search for it (use json format to inspect structured results) response = await search_notes( - project=test_project.name, query="Search Note", search_type="title" + project=test_project.name, query="Search Note", search_type="title", output_format="json" ) # Verify results - handle both success and error cases - if isinstance(response, str): + if isinstance(response, dict): + # Success case - verify dict response + assert len(response["results"]) > 0 + assert any( + r["permalink"] == f"{test_project.name}/test/test-search-note" + for r in response["results"] + ) + else: # If search failed and returned error message, test should fail with informative message pytest.fail(f"Search failed with error: {response}") - else: - # Success case - verify SearchResponse - assert len(response.results) > 0 - assert any( - r.permalink == f"{test_project.name}/test/test-search-note" for r in response.results - ) @pytest.mark.asyncio @@ -80,19 +88,21 @@ async def test_search_permalink(client, test_project): ) assert result - # Search for it + # Search for it (use json format to inspect structured results) response = await search_notes( project=test_project.name, query=f"{test_project.name}/test/test-search-note", search_type="permalink", + output_format="json", ) # Verify results - handle both success and error cases - if isinstance(response, SearchResponse): - # Success case - verify SearchResponse - assert len(response.results) > 0 + if isinstance(response, dict): + # Success case - verify dict response + assert len(response["results"]) > 0 assert any( - r.permalink == f"{test_project.name}/test/test-search-note" for r in response.results + r["permalink"] == f"{test_project.name}/test/test-search-note" + for r in response["results"] ) else: # If search failed and returned error message, test should fail with informative message @@ -112,19 +122,21 @@ async def test_search_permalink_match(client, test_project): ) assert result - # Search for it + # Search for it (use json format to inspect structured results) response = await search_notes( project=test_project.name, query=f"{test_project.name}/test/test-search-*", search_type="permalink", + output_format="json", ) # Verify results - handle both success and error cases - if isinstance(response, SearchResponse): - # Success case - verify SearchResponse - assert len(response.results) > 0 + if isinstance(response, dict): + # Success case - verify dict response + assert len(response["results"]) > 0 assert any( - r.permalink == f"{test_project.name}/test/test-search-note" for r in response.results + r["permalink"] == f"{test_project.name}/test/test-search-note" + for r in response["results"] ) else: # If search failed and returned error message, test should fail with informative message @@ -142,13 +154,15 @@ async def test_search_memory_url_with_project_prefix(client, test_project): ) assert result - response = await search_notes(query=f"memory://{test_project.name}/test/memory-url-search-note") + response = await search_notes( + query=f"memory://{test_project.name}/test/memory-url-search-note", output_format="json" + ) - if isinstance(response, SearchResponse): - assert len(response.results) > 0 + if isinstance(response, dict): + assert len(response["results"]) > 0 assert any( - r.permalink == f"{test_project.name}/test/memory-url-search-note" - for r in response.results + r["permalink"] == f"{test_project.name}/test/memory-url-search-note" + for r in response["results"] ) else: pytest.fail(f"Search failed with error: {response}") @@ -167,17 +181,18 @@ async def test_search_pagination(client, test_project): ) assert result - # Search for it + # Search for it (use json format to inspect structured results) response = await search_notes( - project=test_project.name, query="searchable", page=1, page_size=1 + project=test_project.name, query="searchable", page=1, page_size=1, output_format="json" ) # Verify results - handle both success and error cases - if isinstance(response, SearchResponse): - # Success case - verify SearchResponse - assert len(response.results) == 1 + if isinstance(response, dict): + # Success case - verify dict response + assert len(response["results"]) == 1 assert any( - r.permalink == f"{test_project.name}/test/test-search-note" for r in response.results + r["permalink"] == f"{test_project.name}/test/test-search-note" + for r in response["results"] ) else: # If search failed and returned error message, test should fail with informative message @@ -195,13 +210,15 @@ async def test_search_with_type_filter(client, test_project): content="# Test\nFiltered by type", ) - # Search with note type filter - response = await search_notes(project=test_project.name, query="type", note_types=["note"]) + # Search with note type filter (use json format to inspect structured results) + response = await search_notes( + project=test_project.name, query="type", note_types=["note"], output_format="json" + ) # Verify results - handle both success and error cases - if isinstance(response, SearchResponse): + if isinstance(response, dict): # Success case - verify all results are entities - assert all(r.type == "entity" for r in response.results) + assert all(r["type"] == "entity" for r in response["results"]) else: # If search failed and returned error message, test should fail with informative message pytest.fail(f"Search failed with error: {response}") @@ -218,13 +235,15 @@ async def test_search_with_entity_type_filter(client, test_project): content="# Test\nFiltered by type", ) - # Search with entity_types (SearchItemType) filter - response = await search_notes(project=test_project.name, query="type", entity_types=["entity"]) + # Search with entity_types (SearchItemType) filter (use json format) + response = await search_notes( + project=test_project.name, query="type", entity_types=["entity"], output_format="json" + ) # Verify results - handle both success and error cases - if isinstance(response, SearchResponse): + if isinstance(response, dict): # Success case - verify all results are entities - assert all(r.type == "entity" for r in response.results) + assert all(r["type"] == "entity" for r in response["results"]) else: # If search failed and returned error message, test should fail with informative message pytest.fail(f"Search failed with error: {response}") @@ -241,16 +260,19 @@ async def test_search_with_date_filter(client, test_project): content="# Test\nRecent content", ) - # Search with date filter + # Search with date filter (use json format to inspect structured results) one_hour_ago = datetime.now() - timedelta(hours=1) response = await search_notes( - project=test_project.name, query="recent", after_date=one_hour_ago.isoformat() + project=test_project.name, + query="recent", + after_date=one_hour_ago.isoformat(), + output_format="json", ) # Verify results - handle both success and error cases - if isinstance(response, SearchResponse): + if isinstance(response, dict): # Success case - verify we get results within timeframe - assert len(response.results) > 0 + assert len(response["results"]) > 0 else: # If search failed and returned error message, test should fail with informative message pytest.fail(f"Search failed with error: {response}") @@ -468,7 +490,8 @@ async def test_search_notes_sets_retrieval_mode_for_semantic_types(monkeypatch, search_type=search_type, ) - assert isinstance(result, SearchResponse) + # Default text format returns a formatted string for empty results + assert isinstance(result, str) assert captured_payload["text"] == "semantic lookup" # "semantic" is an alias for "vector" retrieval mode expected_mode = "vector" if search_type in ("vector", "semantic") else search_type @@ -563,7 +586,8 @@ async def test_search_notes_filter_only_metadata(monkeypatch): metadata_filters={"status": "in-progress"}, ) - assert isinstance(result, SearchResponse) + # Default text format returns a formatted string for empty results + assert isinstance(result, str) assert captured_payload["metadata_filters"] == {"status": "in-progress"} # No text/title/permalink should be set assert captured_payload.get("text") is None @@ -605,7 +629,8 @@ async def test_search_notes_filter_only_tags(monkeypatch): tags=["security", "oauth"], ) - assert isinstance(result, SearchResponse) + # Default text format returns a formatted string for empty results + assert isinstance(result, str) assert captured_payload["tags"] == ["security", "oauth"] assert captured_payload.get("text") is None @@ -1158,7 +1183,8 @@ async def test_search_notes_tag_prefix_converts_to_tags_filter(monkeypatch): query="tag:security", ) - assert isinstance(result, SearchResponse) + # Default text format returns a formatted string for empty results + assert isinstance(result, str) assert captured_payload["tags"] == ["security"] # No text query should be set — tag: prefix was consumed assert captured_payload.get("text") is None @@ -1199,7 +1225,8 @@ async def test_search_notes_tag_prefix_merges_with_explicit_tags(monkeypatch): tags=["oauth"], ) - assert isinstance(result, SearchResponse) + # Default text format returns a formatted string for empty results + assert isinstance(result, str) assert set(captured_payload["tags"]) == {"security", "oauth"} assert captured_payload.get("text") is None @@ -1238,7 +1265,8 @@ async def test_search_notes_multiple_tag_prefixes(monkeypatch): query="tag:coffee AND tag:brewing", ) - assert isinstance(result, SearchResponse) + # Default text format returns a formatted string for empty results + assert isinstance(result, str) assert set(captured_payload["tags"]) == {"coffee", "brewing"} # Boolean connector AND should be stripped, leaving no text query assert captured_payload.get("text") is None @@ -1283,7 +1311,211 @@ async def test_search_notes_tag_prefix_with_remaining_text(monkeypatch): query="authentication tag:security", ) - assert isinstance(result, SearchResponse) + # Default text format returns a formatted string for empty results + assert isinstance(result, str) assert captured_payload["tags"] == ["security"] # Remaining text should be preserved as the query assert captured_payload["text"] == "authentication" + + +# --- Tests for text output format (#641) ----------------------------------- + + +def test_format_search_markdown_with_results(): + """_format_search_markdown returns readable markdown for non-empty results.""" + from basic_memory.schemas.search import SearchResult, SearchItemType + + result = SearchResponse( + results=[ + SearchResult( + title="My Note", + type=SearchItemType.ENTITY, + score=0.85, + permalink="docs/my-note", + file_path="docs/My Note.md", + matched_chunk="This is a matching snippet", + ), + SearchResult( + title="Other Note", + type=SearchItemType.ENTITY, + score=0.42, + permalink="docs/other-note", + file_path="docs/Other Note.md", + ), + ], + current_page=1, + page_size=10, + ) + + text = _format_search_markdown(result, "test-project", "my query") + assert isinstance(text, str) + assert "# Search Results: my query" in text + assert "test-project" in text + assert "### My Note" in text + assert "permalink: docs/my-note" in text + assert "0.8500" in text + assert "match: This is a matching snippet" in text + assert "### Other Note" in text + assert "2 results" in text + assert "page 1" in text + + +def test_format_search_markdown_empty_results(): + """_format_search_markdown returns a no-results message when results are empty.""" + result = SearchResponse(results=[], current_page=1, page_size=10) + text = _format_search_markdown(result, "test-project", "missing") + assert isinstance(text, str) + assert "No results found" in text + assert "missing" in text + + +@pytest.mark.asyncio +async def test_search_notes_text_format_returns_string(monkeypatch): + """search_notes with output_format='text' returns a formatted markdown string.""" + import importlib + + from basic_memory.schemas.search import SearchResult, SearchItemType + + search_mod = importlib.import_module("basic_memory.mcp.tools.search") + clients_mod = importlib.import_module("basic_memory.mcp.clients") + + class StubProject: + name = "test-project" + external_id = "test-external-id" + + @asynccontextmanager + async def fake_get_project_client(*args, **kwargs): + yield (object(), StubProject()) + + async def fake_resolve_project_and_path( + client, identifier, project=None, context=None, headers=None + ): + return StubProject(), identifier, False + + class MockSearchClient: + def __init__(self, *args, **kwargs): + pass + + async def search(self, payload, page, page_size): + return SearchResponse( + results=[ + SearchResult( + title="Found Note", + type=SearchItemType.ENTITY, + score=0.9, + permalink="docs/found-note", + file_path="docs/Found Note.md", + matched_chunk="snippet", + ), + ], + current_page=page, + page_size=page_size, + ) + + monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client) + monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path) + monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient) + + result = await search_mod.search_notes( + project="test-project", + query="test", + output_format="text", + ) + + assert isinstance(result, str) + assert "# Search Results: test" in result + assert "### Found Note" in result + assert "permalink: docs/found-note" in result + + +# --- Tests for metadata_filters key aliasing (#642) ---------------------------- + + +@pytest.mark.asyncio +async def test_search_notes_metadata_filters_aliases_note_type(monkeypatch): + """metadata_filters={'note_type': 'note'} is aliased to {'type': 'note'}.""" + import importlib + + search_mod = importlib.import_module("basic_memory.mcp.tools.search") + clients_mod = importlib.import_module("basic_memory.mcp.clients") + + class StubProject: + name = "test-project" + external_id = "test-external-id" + + @asynccontextmanager + async def fake_get_project_client(*args, **kwargs): + yield (object(), StubProject()) + + async def fake_resolve_project_and_path( + client, identifier, project=None, context=None, headers=None + ): + return StubProject(), identifier, False + + captured_payload: dict = {} + + class MockSearchClient: + def __init__(self, *args, **kwargs): + pass + + async def search(self, payload, page, page_size): + captured_payload.update(payload) + return SearchResponse(results=[], current_page=page, page_size=page_size) + + monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client) + monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path) + monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient) + + await search_mod.search_notes( + project="test-project", + query="test", + metadata_filters={"note_type": "note"}, + ) + + # "note_type" should be aliased to "type" in the payload + assert captured_payload["metadata_filters"] == {"type": "note"} + + +@pytest.mark.asyncio +async def test_search_notes_metadata_filters_preserves_non_aliased_keys(monkeypatch): + """metadata_filters with non-aliased keys pass through unchanged.""" + import importlib + + search_mod = importlib.import_module("basic_memory.mcp.tools.search") + clients_mod = importlib.import_module("basic_memory.mcp.clients") + + class StubProject: + name = "test-project" + external_id = "test-external-id" + + @asynccontextmanager + async def fake_get_project_client(*args, **kwargs): + yield (object(), StubProject()) + + async def fake_resolve_project_and_path( + client, identifier, project=None, context=None, headers=None + ): + return StubProject(), identifier, False + + captured_payload: dict = {} + + class MockSearchClient: + def __init__(self, *args, **kwargs): + pass + + async def search(self, payload, page, page_size): + captured_payload.update(payload) + return SearchResponse(results=[], current_page=page, page_size=page_size) + + monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client) + monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path) + monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient) + + await search_mod.search_notes( + project="test-project", + query="test", + metadata_filters={"note_type": "spec", "priority": "high"}, + ) + + # "note_type" aliased to "type", "priority" passes through unchanged + assert captured_payload["metadata_filters"] == {"type": "spec", "priority": "high"} diff --git a/tests/services/test_link_resolver.py b/tests/services/test_link_resolver.py index 38ba44bb..bb91656a 100644 --- a/tests/services/test_link_resolver.py +++ b/tests/services/test_link_resolver.py @@ -901,3 +901,21 @@ async def test_resolve_link_non_uuid_falls_through(link_resolver, test_entities, result = await link_resolver.resolve_link("Core Service") assert result is not None assert result.permalink == f"{project_prefix}/components/core-service" + + +# ============================================================================ +# Fuzzy search best-match selection tests (#640) +# ============================================================================ + + +@pytest.mark.asyncio +async def test_fuzzy_search_selects_first_result(link_resolver, project_prefix): + """Test that fuzzy search uses results[0] (best-ranked by the DB) regardless of score sign. + + Both SQLite (BM25, negative scores, ASC) and Postgres (ts_rank, positive scores, DESC) + return the best match first. Using results[0] is backend-agnostic and correct. + """ + result = await link_resolver.resolve_link("Auth Serv") + assert result is not None + # The best match for "Auth Serv" should be Auth Service + assert result.permalink == f"{project_prefix}/components/auth-service"