feat(search): add graceful degradation with fallback for ha_search_entities (#231)

Implements issue #214: When the primary fuzzy search fails, the search
now attempts fallback methods before returning an error:

1. Try fuzzy search (primary method)
2. If that fails, try exact substring matching
3. If that fails, return partial results listing

Response now includes `partial: true` and `warning` fields when fallback
methods are used, matching the format specified in the issue.

Closes #214

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Claude <noreply@anthropic.com>

Author: julienld

src/ha_mcp/tools/tools_search.py

@@ -4,13 +4,100 @@
 This module provides entity search, system overview, deep search, and state retrieval tools.
 """
 
+import logging
 from typing import Annotated, Any, Literal, cast
 
 from pydantic import Field
 
 from .helpers import log_tool_usage
 from .util_helpers import add_timezone_metadata, coerce_bool_param, parse_string_list_param
 
+logger = logging.getLogger(__name__)
+
+
+async def _exact_match_search(
+    client, query: str, domain_filter: str | None, limit: int
+) -> dict[str, Any]:
+    """
+    Fallback exact match search when fuzzy search fails.
+
+    Performs simple substring matching on entity_id and friendly_name.
+    """
+    all_entities = await client.get_states()
+    query_lower = query.lower().strip()
+
+    results = []
+    for entity in all_entities:
+        entity_id = entity.get("entity_id", "")
+        attributes = entity.get("attributes", {})
+        friendly_name = attributes.get("friendly_name", entity_id)
+        domain = entity_id.split(".")[0] if "." in entity_id else ""
+
+        # Apply domain filter if provided
+        if domain_filter and domain != domain_filter:
+            continue
+
+        # Check for exact substring match in entity_id or friendly_name
+        if query_lower in entity_id.lower() or query_lower in friendly_name.lower():
+            results.append({
+                "entity_id": entity_id,
+                "friendly_name": friendly_name,
+                "domain": domain,
+                "state": entity.get("state", "unknown"),
+                "score": 100 if query_lower == entity_id.lower() or query_lower == friendly_name.lower() else 80,
+                "match_type": "exact_match",
+            })
+
+    # Sort by score descending
+    results.sort(key=lambda x: x["score"], reverse=True)
+    return {
+        "success": True,
+        "query": query,
+        "total_matches": len(results),
+        "results": results[:limit],
+        "search_type": "exact_match",
+    }
+
+
+async def _partial_results_search(
+    client, query: str, domain_filter: str | None, limit: int
+) -> dict[str, Any]:
+    """
+    Last resort fallback - return any entities that might be relevant.
+
+    Returns entities from the specified domain (if any) or a sample of all entities.
+    """
+    all_entities = await client.get_states()
+
+    results = []
+    for entity in all_entities:
+        entity_id = entity.get("entity_id", "")
+        attributes = entity.get("attributes", {})
+        friendly_name = attributes.get("friendly_name", entity_id)
+        domain = entity_id.split(".")[0] if "." in entity_id else ""
+
+        # Apply domain filter if provided
+        if domain_filter and domain != domain_filter:
+            continue
+
+        results.append({
+            "entity_id": entity_id,
+            "friendly_name": friendly_name,
+            "domain": domain,
+            "state": entity.get("state", "unknown"),
+            "score": 0,  # No match score for partial results
+            "match_type": "partial_listing",
+        })
+
+    return {
+        "success": True,
+        "partial": True,
+        "query": query,
+        "total_matches": len(results),
+        "results": results[:limit],
+        "search_type": "partial_listing",
+    }
+
 
 def register_search_tools(mcp, client, **kwargs):
     """Register search and discovery tools with the MCP server."""
@@ -211,14 +298,50 @@ async def ha_search_entities(
                     domain_list_data["by_domain"] = {domain_filter: results}
                 return await add_timezone_metadata(client, domain_list_data)
 
-            result = await smart_tools.smart_entity_search(query, limit)
+            # Graceful degradation with fallback search methods
+            # 1. Try fuzzy search (primary method)
+            # 2. If that fails, try exact match
+            # 3. If that fails, return partial results with warning
+            # 4. Only error if all methods fail
+
+            result = None
+            warning = None
+            search_type = "fuzzy_search"
+
+            # Step 1: Try fuzzy search
+            try:
+                result = await smart_tools.smart_entity_search(query, limit)
+                search_type = "fuzzy_search"
+            except Exception as fuzzy_error:
+                logger.warning(f"Fuzzy search failed, trying exact match: {fuzzy_error}")
+
+                # Step 2: Try exact match fallback
+                try:
+                    result = await _exact_match_search(client, query, domain_filter, limit)
+                    warning = "Fuzzy search unavailable, using exact match"
+                    search_type = "exact_match"
+                except Exception as exact_error:
+                    logger.warning(f"Exact match failed, trying partial results: {exact_error}")
+
+                    # Step 3: Try partial results fallback
+                    try:
+                        result = await _partial_results_search(client, query, domain_filter, limit)
+                        warning = "Search degraded, returning partial results"
+                        search_type = "partial_listing"
+                    except Exception as partial_error:
+                        # Step 4: All methods failed - raise to outer exception handler
+                        logger.error(f"All search methods failed: {partial_error}")
+                        raise Exception(
+                            f"All search methods failed. Fuzzy: {fuzzy_error}, "
+                            f"Exact: {exact_error}, Partial: {partial_error}"
+                        ) from partial_error
 
             # Convert 'matches' to 'results' for backward compatibility
             if "matches" in result:
                 result["results"] = result.pop("matches")
 
-            # Apply domain filter if provided
-            if domain_filter and "results" in result:
+            # Apply domain filter if provided (for fuzzy search results)
+            if domain_filter and "results" in result and search_type == "fuzzy_search":
                 filtered_results = [
                     r for r in result["results"] if r.get("domain") == domain_filter
                 ]
@@ -236,7 +359,13 @@ async def ha_search_entities(
                     by_domain[domain].append(entity)
                 result["by_domain"] = by_domain
 
-            result["search_type"] = "fuzzy_search"
+            result["search_type"] = search_type
+
+            # Add warning and partial flag if fallback was used
+            if warning:
+                result["warning"] = warning
+                result["partial"] = True
+
             return await add_timezone_metadata(client, result)
 
         except Exception as e:

tests/src/e2e/tools/test_search_entities.py

@@ -225,3 +225,94 @@ async def test_search_entities_multiple_domains(mcp_client):
     # At least one domain should have results
     assert any(count > 0 for count in results_summary.values()), \
         "Expected at least one domain to have entities"
+
+
+# ============================================================================
+# Tests for graceful degradation (issue #214)
+# ============================================================================
+
+
+@pytest.mark.asyncio
+async def test_search_entities_successful_fuzzy_search_no_warning(mcp_client):
+    """Test that successful fuzzy search returns no warning or partial flag.
+
+    Issue #214: Normal fuzzy search should work without fallback indicators.
+    """
+    logger.info("Testing successful fuzzy search has no fallback indicators")
+
+    result = await mcp_client.call_tool(
+        "ha_search_entities",
+        {"query": "light", "limit": 5},
+    )
+    raw_data = assert_mcp_success(result, "Fuzzy search success")
+    data = raw_data.get("data", raw_data)
+
+    assert data.get("success") is True
+    assert data.get("search_type") == "fuzzy_search"
+    # Normal search should NOT have warning or partial flag
+    assert "warning" not in data or data.get("warning") is None
+    assert "partial" not in data or data.get("partial") is not True
+
+    logger.info("Fuzzy search succeeded without fallback indicators")
+
+
+@pytest.mark.asyncio
+async def test_search_entities_response_structure_issue_214(mcp_client):
+    """Test that search response has the expected structure from issue #214.
+
+    The response should include:
+    - success: boolean
+    - results: array
+    - search_type: string indicating which method was used
+    """
+    logger.info("Testing response structure for issue #214")
+
+    result = await mcp_client.call_tool(
+        "ha_search_entities",
+        {"query": "light", "limit": 5},
+    )
+    raw_data = assert_mcp_success(result, "Response structure check")
+    data = raw_data.get("data", raw_data)
+
+    # Verify required fields
+    assert "success" in data, "Response must include 'success' field"
+    assert "results" in data, "Response must include 'results' field"
+    assert "search_type" in data, "Response must include 'search_type' field"
+    assert isinstance(data["results"], list), "Results must be a list"
+
+    # search_type should be one of the expected values
+    valid_search_types = ["fuzzy_search", "exact_match", "partial_listing", "domain_listing"]
+    assert data["search_type"] in valid_search_types, \
+        f"search_type '{data['search_type']}' not in {valid_search_types}"
+
+    logger.info(f"Response structure valid with search_type: {data['search_type']}")
+
+
+@pytest.mark.asyncio
+async def test_search_entities_fallback_fields_when_present(mcp_client):
+    """Test that fallback fields have correct types when present.
+
+    Issue #214: When fallback is used, response should include:
+    - partial: true
+    - warning: string explaining what happened
+    """
+    logger.info("Testing fallback field types")
+
+    result = await mcp_client.call_tool(
+        "ha_search_entities",
+        {"query": "light", "limit": 5},
+    )
+    raw_data = assert_mcp_success(result, "Fallback field types")
+    data = raw_data.get("data", raw_data)
+
+    # If warning is present, it should be a string
+    if "warning" in data and data["warning"] is not None:
+        assert isinstance(data["warning"], str), "warning must be a string"
+        logger.info(f"Warning present: {data['warning']}")
+
+    # If partial is present, it should be a boolean
+    if "partial" in data and data["partial"] is not None:
+        assert isinstance(data["partial"], bool), "partial must be a boolean"
+        logger.info(f"Partial flag: {data['partial']}")
+
+    logger.info("Fallback field types are correct")